package com.yl.flink.processor;

import cn.hutool.core.util.StrUtil;
import com.yl.constant.Const;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBPoint;

import java.util.HashMap;

public class CountPointMapFunc implements MapFunction<Tuple4<String, String, Long, Integer>, InfluxDBPoint> {

    /**
     * 把统计的小时数据量组装成influxdb point
     */
    @Override
    public InfluxDBPoint map(Tuple4<String, String, Long, Integer> element) throws Exception {
        // 表名
        String tbl = StrUtil.format(Const.TPL_TBL_COUNT, element.f0.toLowerCase());
        // 数据时间
        long timestamp = element.f2;
        // tags
        HashMap<String, String> tags = new HashMap<>();
        tags.put(Const.S_TID, element.f1);
        // fields
        HashMap<String, Object> fields = new HashMap<>();
        fields.put(Const.S_COUNT, element.f3);
        return new InfluxDBPoint(tbl, timestamp, tags, fields);
    }
}
